package com.atguigu.flink.streamapi.source;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;

/**
 * Created by Smexy on 2022/11/21
 *
 *      流式App，从消息中间件中读取数据，例如kafka
 *
 *      通过 StreamExecutionEnvironment可以直接获取的都是flink自带的Source。
 *
 *      kafkasource不是自带，需要额外安装
 *
 *      核心： 创建 FlinkKafkaConsumer
 *
 *      flink中读取数据源:    env.addSource(xxx)
 *              调用是自带的Source，通常只需要调用env封装好的方法: env.socketTextStream()
 *              调用的是额外安装的Source，需要自己手动 调用 env.addSource
 *
 *   -------------------------------------------
 *      因为：  在flink中消费kafka的偏移量，是默认在算子的状态(保存的数据)中保存。和kafka远程保存的没关系。
 *
 *             FlinkKafkaConsumer在工作时，如果使用了的是kafkaConsumer.setStartFromLatest()即使本地算子的状态中没有保存偏移量，而kafka中保存了偏移量，也不会使用！
 *             如果使用 kafkaConsumer.setStartFromGroupOffsets()，本地算子中没有偏移量，会从远程读取提交的便宜量，向后消费。
 *
 *
 *    -----------------
 *     1.14之后过时
 */
public class Demo3_KafkaSource_Old
{
    public static void main(String[] args) {



        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "hadoop102:9092");
        properties.setProperty("group.id", "sz0704b");
        properties.setProperty("auto.commit.interval.ms","500"); // 每隔多久自动提交一次
        properties.setProperty("enable.auto.commit","true"); //允许自动提交offsets

        /*
                创建一个 FlinkKafkaConsumer
                    通常情况下，是只需要传入Value的反序列化器。根据value的类型传入合适的反序列化器。

                        kafka中存储的数据的结构:
                                   生产ProducerRecord  ----->broker ------> 消费ConsumerRecord
                                    K: 只用于存储元数据。或用于指定分区
                                    V： 真正存储数据
         */
        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>("t1", new SimpleStringSchema(), properties);

        /*
            可以指定消费者的消费策略
                 如果当前组已经消费过了，从上次提交的位置消费。

                 没有消费过，采取默认策略,以下两个参数才有生效的可能。
                        latest: 从最后的位置消费
                        earliest: 最主题最早的位置消费



         */
        kafkaConsumer.setStartFromGroupOffsets();  //从当前组消费过的位置向后消费。 默认
        //kafkaConsumer.setStartFromLatest();   //强制重置到 Latest
        // kafkaConsumer.setStartFromSpecificOffsets();  从指定的 位置消费

        //调用addSource()
        DataStream<String> stream = env.addSource(kafkaConsumer);

        stream.print();

        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }

    }
}
